package queue

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"
	"video-api/internal/consts"
	"video-api/internal/dao"
	"video-api/internal/model/entity"
	"video-api/utility/redis_client"

	"github.com/gogf/gf/v2/container/gqueue"
	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/redis/go-redis/v9"
)

var Gq *gqueue.Queue

type Task struct {
	QueueName string
	Data      interface{}
}

func Init(ctx context.Context) {
	if Gq != nil {
		g.Log().Error(ctx, "queue already initialized")
		return
	}
	Gq = gqueue.New()
	if Gq == nil {
		g.Log().Fatal(ctx, "failed to create queue")
		return
	}
	go DoTask(ctx)
}

// PushTask 推送任务到队列
func PushTask(ctx context.Context, queueName string, data interface{}) error {
	defer func() { // 防止投递任务主程序崩溃
		if err := recover(); err != nil {
			g.Log().Error(ctx, "PushTask panic recovered:", err)
		}
	}()
	if queueName == "" {
		return fmt.Errorf("queue name cannot be empty")
	}
	if data == nil {
		return fmt.Errorf("task data cannot be nil")
	}
	switch queueName {
	case "add_history_record":
		if _, ok := data.(AddShortDramaHistory); !ok {
			return fmt.Errorf("invalid data type for add_history_record")
		}
	case "del_history_record":
		if _, ok := data.(DelShortDramaHistory); !ok {
			return fmt.Errorf("invalid data type for del_history_record")
		}
	case "exec_sql":
		if _, ok := data.(string); !ok {
			return fmt.Errorf("invalid data type for exec_sql")
		}
	default:
		return fmt.Errorf("unknown queue name: %s", queueName)
	}
	task := Task{
		QueueName: queueName,
		Data:      data,
	}
	if Gq == nil {
		return fmt.Errorf("queue not initialized")
	}
	// g.Log().Debug(ctx, "pushing task to queue", "queue", queueName, "data", data)
	Gq.Push(task)
	return nil
}

func DoTask(ctx context.Context) {
	defer func() {
		if err := recover(); err != nil {
			g.Log().Error(ctx, "DoTask panic recovered:", err)
			go DoTask(ctx)
		}
	}()
	for queueItem := range Gq.C {
		task, ok := queueItem.(Task)
		if !ok {
			marshal, err := json.Marshal(queueItem)
			if err != nil {
				g.Log().Error(ctx, "task marshal err:", err)
			} else {
				g.Log().Error(ctx, "task assert err, data:", string(marshal))
			}
			continue
		}
		g.Log().Debug(ctx, "processing task:", task.QueueName)
		// 任务列
		switch task.QueueName {
		case "add_history_record":
			doHistoryRecord(ctx, task)
		case "del_history_record":
			delHistoryRecord(ctx, task)
		case "exec_sql":
			execSql(ctx, task)
		default:
			g.Log().Error(ctx, "unknown task type:", task.QueueName)
		}
		g.Log().Debug(ctx, "task completed:", task.QueueName)
	}
}

// 添加短剧浏览历史
func doHistoryRecord(ctx context.Context, task Task) {
	// 添加超时
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if data, ok := task.Data.(AddShortDramaHistory); !ok {
		marshal, err := json.Marshal(task.Data)
		if err != nil {
			g.Log().Error(ctx, "task child marshal err.", err)
			return
		}
		g.Log().Error(ctx, "task child assert err.", "data:", marshal)
		return
	} else {
		// 待定，感觉暂时可有可无？
		// lockKey := fmt.Sprintf("lock:history:%d:%d", data.UserId, data.ShortDramaId)
		// lock := redis_client.RedisClient.SetNX(ctx, lockKey, "1", 10*time.Second)
		// if !lock.Val() {
		// 	g.Log().Warning(ctx, "doHistoryRecord locked, skip processing")
		// 	return
		// }
		// defer redis_client.RedisClient.Del(ctx, lockKey)
		// 事务
		err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
			// 查询现有记录
			sdrModel := dao.ShortDramaRecord.Ctx(ctx)
			sdrResult, err := sdrModel.Where("user_id = ?", data.UserId).
				Where("short_drama_id = ?", data.ShortDramaId).One()
			if err != nil && errors.Is(err, sql.ErrNoRows) {
				return fmt.Errorf("query history record error: %w", err)
			}
			// 获取短剧信息
			sdModel := dao.ShortDrama.Ctx(ctx)
			shortDramaInfo, err := sdModel.Where("id = ?", data.ShortDramaId).
				Fields("id as short_drama_id,short_name,description,cover_img,tags").One()
			if err != nil {
				return fmt.Errorf("query short drama error: %w", err)
			}
			if len(shortDramaInfo) == 0 {
				return fmt.Errorf("short drama not found: %d", data.ShortDramaId)
			}
			// 获取标签信息
			tags := make([]entity.Tags, 0)
			sdtModel := dao.ShortDramaTags.Ctx(ctx).Fields("t.*")
			err = sdtModel.LeftJoin("tags t", "t.id = short_drama_tags.tag_id").
				//Where("short_drama_tags.tag_id = ?", shortDramaInfo["tags"]).
				Where("short_drama_id = ?", data.ShortDramaId).Scan(&tags)
			if err != nil && err != sql.ErrNoRows {
				return fmt.Errorf("query tags error: %w", err)
			}
			// 获取剧集信息
			sdeModel := dao.ShortDramaEpisodes.Ctx(ctx)
			shortDramaEpisodesInfo, err := sdeModel.Where("id = ?", data.EpisodesId).
				Fields("order").One()
			if err != nil {
				return fmt.Errorf("query episodes error: %w", err)
			}
			if len(shortDramaEpisodesInfo) == 0 {
				return fmt.Errorf("episode not found: %d", data.EpisodesId)
			}
			var historyId int64
			tagsList := make([]string, 0)
			if len(tags) > 0 {
				for _, tag := range tags {
					tagsList = append(tagsList, tag.Title)
				}
			}
			//tageByte, _ := json.Marshal(tagsList)
			insertOrUpdateData := g.Map{
				"user_id":        data.UserId,
				"short_drama_id": data.ShortDramaId,
				"episodes_id":    data.EpisodesId,
				"play_at":        data.PlayTime,
				"category":       consts.HISTORY,
				"order":          shortDramaEpisodesInfo["order"],
				"short_name":     shortDramaInfo["short_name"],
				"description":    shortDramaInfo["description"],
				"cover_img":      shortDramaInfo["cover_img"],
				"tags":           strings.Join(tagsList, ","),
			}
			if len(sdrResult) == 0 {
				insertResult, err := sdrModel.Data(insertOrUpdateData).Insert()
				if err != nil {
					return fmt.Errorf("insert history record error: %w", err)
				}
				historyId, err = insertResult.LastInsertId()
				if err != nil {
					return fmt.Errorf("get last insert id error: %w", err)
				}
			} else {
				historyId = sdrResult["id"].Int64()
				_, err := sdrModel.Data(insertOrUpdateData).
					Where("user_id = ?", data.UserId).
					Where("short_drama_id = ?", data.ShortDramaId).
					Update()
				if err != nil {
					return fmt.Errorf("update history record error: %w", err)
				}
			}

			// 更新 Redis 缓存
			if err := updateRedisHistory(ctx, insertOrUpdateData, historyId, sdrResult, data); err != nil {
				return fmt.Errorf("update redis error: %w", err)
			}
			return nil
		})
		if err != nil {
			g.Log().Error(ctx, "doHistoryRecord transaction failed:", err)
			return
		}
	}
}

// 更新Redis中的历史记录
//func updateRedisHistory(ctx context.Context, data AddShortDramaHistory, historyId int64,
//	shortDramaInfo, shortDramaEpisodesInfo, sdrResult gdb.Record, tags []string) error {
//	historyData := g.Map{
//		"id":           historyId,
//		"shortDramaId": shortDramaInfo["short_drama_id"],
//		"shortName":    shortDramaInfo["short_name"],
//		"description":  shortDramaInfo["description"],
//		"coverImg":     shortDramaInfo["cover_img"],
//		"tags":         tags,
//		"order":        shortDramaEpisodesInfo["order"],
//	}
//	key := consts.SHORT_DRAMA_HISTORY + ":" + strconv.Itoa(int(data.UserId))
//	if len(sdrResult) > 0 {
//		oldHistoryData := g.Map{
//			"id":           sdrResult["id"],
//			"shortDramaId": sdrResult["short_drama_id"],
//			"shortName":    shortDramaInfo["short_name"],
//			"description":  shortDramaInfo["description"],
//			"coverImg":     shortDramaInfo["cover_img"],
//			"tags":         tags,
//			"order":        sdrResult["order"],
//		}
//		m, err := json.Marshal(oldHistoryData)
//		if err != nil {
//			return fmt.Errorf("marshal old history data error: %w", err)
//		}
//		if err := redis_client.RedisClient.ZRem(ctx, key, m).Err(); err != nil {
//			return fmt.Errorf("remove old history from redis error: %w", err)
//		}
//	}
//	m, err := json.Marshal(historyData)
//	if err != nil {
//		return fmt.Errorf("marshal new history data error: %w", err)
//	}
//	member := redis.Z{
//		Score:  float64(data.PlayTime.Unix()),
//		Member: m,
//	}
//	if err := redis_client.RedisClient.ZAdd(ctx, key, member).Err(); err != nil {
//		return fmt.Errorf("add new history to redis error: %w", err)
//	}
//	result2, err := redis_client.RedisClient.ZCard(ctx, key).Result()
//	if err != nil {
//		return fmt.Errorf("get redis zcard error: %w", err)
//	}
//	if result2 > int64(100) {
//		if _, err := redis_client.RedisClient.ZRemRangeByRank(ctx, key, 0, result2-int64(100)-1).Result(); err != nil {
//			return fmt.Errorf("trim redis history error: %w", err)
//		}
//	}
//	return nil
//}

// 更新Redis中的历史记录
func updateRedisHistory(ctx context.Context, record g.Map, historyId int64, sdrResult gdb.Record, data AddShortDramaHistory) error {
	historyData := g.Map{
		"id":           historyId,
		"shortDramaId": record["short_drama_id"],
		"shortName":    record["short_name"],
		"description":  record["description"],
		"coverImg":     record["cover_img"],
		"tags":         record["tags"],
		"order":        record["order"],
		"episodesId":   record["episodes_id"],
	}
	//userId, err := strconv.Atoi(record["user_id"].(string))
	//if err != nil {
	//	g.Log().Error(ctx, "updateRedisHistory userId convert to int error:", err)
	//	return err
	//}
	userId, ok := record["user_id"].(int64)
	if !ok {
		return fmt.Errorf("updateRedisHistory user id not int")
	}
	key := consts.SHORT_DRAMA_HISTORY + ":" + strconv.Itoa(int(userId))
	if len(sdrResult) > 0 {
		oldHistoryData := g.Map{
			"id":           sdrResult["id"],
			"shortDramaId": sdrResult["short_drama_id"],
			"shortName":    sdrResult["short_name"],
			"description":  sdrResult["description"],
			"coverImg":     sdrResult["cover_img"],
			"tags":         sdrResult["tags"],
			"order":        sdrResult["order"],
			"episodesId":   sdrResult["episodes_id"],
		}
		m, err := json.Marshal(oldHistoryData)
		if err != nil {
			return fmt.Errorf("marshal old history data error: %w", err)
		}
		if err := redis_client.RedisClient.ZRem(ctx, key, m).Err(); err != nil {
			return fmt.Errorf("remove old history from redis error: %w", err)
		}
	}
	m, err := json.Marshal(historyData)
	if err != nil {
		return fmt.Errorf("marshal new history data error: %w", err)
	}
	member := redis.Z{
		Score:  float64(data.PlayTime.Unix()),
		Member: m,
	}
	if err := redis_client.RedisClient.ZAdd(ctx, key, member).Err(); err != nil {
		return fmt.Errorf("add new history to redis error: %w", err)
	}
	result2, err := redis_client.RedisClient.ZCard(ctx, key).Result()
	if err != nil {
		return fmt.Errorf("get redis zcard error: %w", err)
	}
	if result2 > int64(100) {
		if _, err := redis_client.RedisClient.ZRemRangeByRank(ctx, key, 0, result2-int64(100)-1).Result(); err != nil {
			return fmt.Errorf("trim redis history error: %w", err)
		}
	}
	return nil
}

// 删除短剧浏览历史
func delHistoryRecord(ctx context.Context, task Task) {
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	if data, ok := task.Data.(DelShortDramaHistory); !ok {
		marshal, err := json.Marshal(task.Data)
		if err != nil {
			g.Log().Error(ctx, "task child marshal err.", err)
			return
		}
		g.Log().Error(ctx, "task child assert err.", "data:", marshal)
		return
	} else {
		// lockKey := fmt.Sprintf("lock:del_history:%d", data.RecordId)
		// lock := redis_client.RedisClient.SetNX(ctx, lockKey, "1", 10*time.Second)
		// if !lock.Val() {
		// 	g.Log().Warning(ctx, "delHistoryRecord locked, skip processing")
		// 	return
		// }
		// defer redis_client.RedisClient.Del(ctx, lockKey)
		err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
			sdrModel := dao.ShortDramaRecord.Ctx(ctx)
			res, err := sdrModel.Where("id = ?", data.RecordId).
				Fields("id,short_drama_id,episodes_id,play_at,short_name,description,cover_img,tags,order").One()
			if err != nil {
				return fmt.Errorf("query record error: %w", err)
			}
			if len(res) == 0 {
				return fmt.Errorf("record not exist: %d", data.RecordId)
			}
			// 软删除mysql
			result, err := sdrModel.Where("id = ?", data.RecordId).Delete()
			if err != nil {
				return fmt.Errorf("delete record error: %w", err)
			}
			affected, err := result.RowsAffected()
			if err != nil {
				return fmt.Errorf("get affected rows error: %w", err)
			}
			if affected == 0 {
				return fmt.Errorf("delete failed, no rows affected")
			}
			// 删除 Redis 缓存
			if err = deleteRedisHistory(ctx, data.UserId, res); err != nil {
				return fmt.Errorf("delete redis history error: %w", err)
			}
			return nil
		})
		if err != nil {
			g.Log().Error(ctx, "delHistoryRecord transaction failed:", err)
			return
		}
	}
}

// 删除Redis中的历史记录
func deleteRedisHistory(ctx context.Context, userId int64, record gdb.Record) error {
	//sdModel := dao.ShortDrama.Ctx(ctx)
	//shortDramaInfo, err := sdModel.Where("id = ?", record["short_drama_id"]).
	//	Fields("id as short_drama_id,short_name,description,cover_img,tags").One()
	//if err != nil {
	//	return fmt.Errorf("query short drama error: %w", err)
	//}
	////sdtModel := dao.ShortDramaTags.Ctx(ctx)
	////tags, err := sdtModel.LeftJoin("tags t", "t.id = short_drama_tags.tag_id").
	////	Where("short_drama_tags.tag_id = ?", shortDramaInfo["tags"]).
	////	Where("short_drama_id = ?", record["short_drama_id"]).One()
	//tags := make([]entity.Tags, 0)
	//sdtModel := dao.ShortDramaTags.Ctx(ctx).Fields("t.*")
	//err = sdtModel.LeftJoin("tags t", "t.id = short_drama_tags.tag_id").
	//	//Where("short_drama_tags.tag_id = ?", shortDramaInfo["tags"]).
	//	Where("short_drama_id = ?", record["short_drama_id"]).Scan(&tags)
	//if err != nil && errors.Is(err, sql.ErrNoRows) {
	//	return fmt.Errorf("query tags error: %w", err)
	//}
	//sdeModel := dao.ShortDramaEpisodes.Ctx(ctx)
	//episodesInfo, err := sdeModel.Where("id = ?", record["episodes_id"]).
	//	Fields("order").One()
	//if err != nil {
	//	return fmt.Errorf("query episodes error: %w", err)
	//}
	//tagsList := make([]string, 0)
	//if len(tags) > 0 {
	//	for _, tag := range tags {
	//		tagsList = append(tagsList, tag.Title)
	//	}
	//}
	historyData := g.Map{
		"id":           record["id"],
		"shortDramaId": record["short_drama_id"],
		"shortName":    record["short_name"],
		"description":  record["description"],
		"coverImg":     record["cover_img"],
		"tags":         record["tags"],
		"order":        record["order"],
		"episodesId":   record["episodes_id"],
	}
	m, err := json.Marshal(historyData)
	if err != nil {
		return fmt.Errorf("marshal history data error: %w", err)
	}
	key := consts.SHORT_DRAMA_HISTORY + ":" + strconv.Itoa(int(userId))
	if err = redis_client.RedisClient.ZRem(ctx, key, m).Err(); err != nil {
		fmt.Println("124324", err.Error())
		return fmt.Errorf("remove from redis error: %w", err)
	}
	return nil
}

// sql操作
func execSql(ctx context.Context, task Task) {
	if data, ok := task.Data.(string); ok {
		_, err := g.DB().Raw(data).All()
		if err != nil {
			g.Log().Error(ctx, "exec sql error.", err)
		}
		return
	}
	g.Log().Error(ctx, "exec sql error.", "sql err")
}
